struct MappedDataFrame
    model::DataFrame
    afunction::Function
    grouped_data_frame::GroupedDataFrame
end

realize(subframe) = subframe.parent[subframe.rows, :]

"""
    @group table *by* columns

Note that this will sometimes cause duplication of the grouping variables.
Copies of the grouping values will get stored, but also, grouping variables are
available in subframes.
"""
group(table, conjunction, columns...) = begin
    grouped = DataFrames.groupby(table, [map(get_expression, columns)...])
    MappedDataFrame(realize(grouped[1]), identity, grouped)
end

# TODO: skip evaluation on first group
"""
    @ungroup table
"""
ungroup(m::MappedDataFrame) = DataFrames.combine( map(m.afunction, m.grouped_data_frame) )

apply(afunction, any) = afunction(any)

apply(afunction, m::MappedDataFrame) =
    MappedDataFrame(afunction(m.model), afunction ∘ m.afunction, m.grouped_data_frame)

column_index(table, table_symbol, index = ImmutableDict{Symbol, Any}()) = begin
    for name in names(table)
        index =
            ImmutableDict(index,
                name => :($table_symbol[$(Meta.quot(name))]::$(typeof(table[name]))))
    end
    index
end
column_index(m::MappedDataFrame, table_symbol) = column_index(m.model, table_symbol)

is_negative(x) = x < zero(x)

get_numbers(table, first_selection, other_selections) = begin
    together = Expr(:tuple, get_expression.( (first_selection, other_selections...) )...)
    numbered_index = zip(names(table), 1:length(table))
    result = WithContext(first_selection;
        expression = together,
        environment = merge!(get_environment(first_selection), numbered_index)) |>
        evaluate
    numbers = vcat(result...)
    if all(is_negative, numbers)
        setdiff(1:length(table), -numbers)
    elseif any(is_negative, numbers)
        error("Cannot mix positive and negative indices")
    else
        numbers
    end
end

"""
    @with table with_context

```jldoctest
julia> using LazyQuery

julia> @new_environment;

julia> @chain @evaluate begin
            using LazyQuery
            DataFrame(a = 1, b = 2)
            @with it a + b
        end
1-element DataArrays.DataArray{Int64,1}:
 3
```
"""
with(table, with_context, table_symbol = gensym()) = begin
    new_env = merge!(get_environment(with_context), column_index(table, table_symbol))
    WithContext(with_context;
        expression = :($apply($table_symbol -> $(get_expression(with_context)), $table)),
        environment = new_env) |>
    evaluate
end

add_to!(table, kwargs) = begin
    for (key, value) in kwargs
        table[key] = value
    end
    table
end

add_to_simple(table, kwargs...) = add_to!(copy(table), kwargs)
add_to_simple(d::DataFrames.SubDataFrame, kwargs...) =
    add_to!(realize(d), kwargs)

make_from_simple(table, kwargs...) = add_to!(typeof(table)(), kwargs)
make_from_simple(d::DataFrames.SubDataFrame, kwargs...) =
    add_to!(DataFrames.DataFrame(), kwargs)

where_simple(table, rows) = table[rows, :]

model_associative(afunction, table, first_assignment, more_assignments, table_symbol = gensym()) = begin
    args = map(make_pair, (first_assignment, more_assignments...))
    with(table,
        WithContext(first_assignment,
            expression = :($afunction($table_symbol, $(args...)))),
        table_symbol)
end

"""
    @make_from table assignments...

Make a new table from an existing table. Can be used on tables and grouped
tables.

```jldoctest
julia> using LazyQuery

julia> @new_environment;

julia> @evaluate begin
            using LazyQuery
            d = DataFrame(a = [1, 1, 2, 2], b = [1, 2, 3, 4])
            @make_from d c = ~b / ¬(sum(a))
        end
4×1 DataFrames.DataFrame
│ Row │ c        │
├─────┼──────────┤
│ 1   │ 0.166667 │
│ 2   │ 0.333333 │
│ 3   │ 0.5      │
│ 4   │ 0.666667 │

julia> @chain @evaluate begin
            @group d by a
            @make_from it c = ~b / ¬(sum(a))
            @ungroup it
        end
4×2 DataFrames.DataFrame
│ Row │ a │ c    │
├─────┼───┼──────┤
│ 1   │ 1 │ 0.5  │
│ 2   │ 1 │ 1.0  │
│ 3   │ 2 │ 0.75 │
│ 4   │ 2 │ 1.0  │
```
"""
make_from(table, first_assignment, more_assignments...) =
    model_associative(make_from_simple, table, first_assignment, more_assignments)

"""
    @add_to table assignments...

```jldoctest
julia> using LazyQuery

julia> @new_environment;

julia> @evaluate begin
            using LazyQuery
            d = DataFrame(a = [1, 1, 2, 2], b = [1, 2, 3, 4])
            @add_to d c = ~b / ¬(sum(a))
        end
4×3 DataFrames.DataFrame
│ Row │ a │ b │ c        │
├─────┼───┼───┼──────────┤
│ 1   │ 1 │ 1 │ 0.166667 │
│ 2   │ 1 │ 2 │ 0.333333 │
│ 3   │ 2 │ 3 │ 0.5      │
│ 4   │ 2 │ 4 │ 0.666667 │

julia> @chain @evaluate begin
            @group d by a
            @add_to it c = ~b / ¬(sum(a))
            @ungroup it
        end
4×4 DataFrames.DataFrame
│ Row │ a │ a_1 │ b │ c    │
├─────┼───┼─────┼───┼──────┤
│ 1   │ 1 │ 1   │ 1 │ 0.5  │
│ 2   │ 1 │ 1   │ 2 │ 1.0  │
│ 3   │ 2 │ 2   │ 3 │ 0.75 │
│ 4   │ 2 │ 2   │ 4 │ 1.0  │

julia> @evaluate @add_to d a / 2
ERROR: Unable to decompose assignment a / 2
[...]
```
"""
add_to(table, first_assignment, more_assignments...) =
    model_associative(add_to_simple, table, first_assignment, more_assignments)

"""
    @rows_from table *wherever* condition
    @rows_from table *at* indices

```jldoctest
julia> using LazyQuery

julia> @new_environment;

julia> @evaluate begin
            using LazyQuery
            d = DataFrame(a = [1, 1, 2, 2], b = [1, 2, 3, 4])
            @rows_from d wherever ~b > ¬(mean(b))
        end
2×2 DataFrames.DataFrame
│ Row │ a │ b │
├─────┼───┼───┤
│ 1   │ 2 │ 3 │
│ 2   │ 2 │ 4 │

julia> @chain @evaluate begin
            @group d by a
            @rows_from it wherever ~b > ¬(mean(b))
            @ungroup it
        end
2×3 DataFrames.DataFrame
│ Row │ a │ a_1 │ b │
├─────┼───┼─────┼───┤
│ 1   │ 1 │ 1   │ 2 │
│ 2   │ 2 │ 2   │ 4 │

julia> @evaluate @rows_from d at [1, 3]
2×2 DataFrames.DataFrame
│ Row │ a │ b │
├─────┼───┼───┤
│ 1   │ 1 │ 1 │
│ 2   │ 2 │ 3 │
```
"""
rows_from(table, at_symbol, rows, table_symbol = gensym()) =
    with(table,
        WithContext(rows,
            expression = :($where_simple($table_symbol, $(maybe_broadcast(get_expression(rows)))))),
        table_symbol)

order_simple(table, rows) = table[sortperm(rows), :]

"""
    @order table *by* rows

```jldoctest
julia> using LazyQuery

julia> @new_environment;

julia> @evaluate begin
            using LazyQuery
            d = DataFrame(a = [1, 1, 2, 2], b = [1, 4, 3, 2])
            @order d by b
        end
4×2 DataFrames.DataFrame
│ Row │ a │ b │
├─────┼───┼───┤
│ 1   │ 1 │ 1 │
│ 2   │ 2 │ 2 │
│ 3   │ 2 │ 3 │
│ 4   │ 1 │ 4 │

julia> @chain @evaluate begin
            @group d by a
            @order it by b
            @ungroup it
        end
4×3 DataFrames.DataFrame
│ Row │ a │ a_1 │ b │
├─────┼───┼─────┼───┤
│ 1   │ 1 │ 1   │ 1 │
│ 2   │ 1 │ 1   │ 4 │
│ 3   │ 2 │ 2   │ 2 │
│ 4   │ 2 │ 2   │ 3 │
```
"""
order(table, by_symbol, rows, table_symbol = gensym()) =
    with(table,
        WithContext(rows,
            expression = :($order_simple($table_symbol, $(maybe_broadcast(get_expression(rows)))))),
        table_symbol)

"""
    @fuze table1 conjunction table2

```jldoctest
julia> using LazyQuery

julia> @new_environment;

julia> @evaluate begin
            using LazyQuery
            df1 = DataFrame(a = [1, 2, 3], b = [4, 5, 6])
            df2 = DataFrame(a = [1, 2, 4], c = [7, 8, 9])
            @fuze df1 and df2
        end
2×3 DataFrames.DataFrame
│ Row │ a │ b │ c │
├─────┼───┼───┼───┤
│ 1   │ 1 │ 4 │ 7 │
│ 2   │ 2 │ 5 │ 8 │

julia> @evaluate @fuze df1 or df2
4×3 DataFrames.DataFrame
│ Row │ a │ b  │ c  │
├─────┼───┼────┼────┤
│ 1   │ 1 │ 4  │ 7  │
│ 2   │ 2 │ 5  │ 8  │
│ 3   │ 3 │ 6  │ NA │
│ 4   │ 4 │ NA │ 9  │

julia> @evaluate @fuze df2 into df1
3×3 DataFrames.DataFrame
│ Row │ c  │ a │ b │
├─────┼────┼───┼───┤
│ 1   │ 7  │ 1 │ 4 │
│ 2   │ 8  │ 2 │ 5 │
│ 3   │ NA │ 3 │ 6 │

julia> @evaluate @fuze df1 with df2
2×2 DataFrames.DataFrame
│ Row │ a │ b │
├─────┼───┼───┤
│ 1   │ 1 │ 4 │
│ 2   │ 2 │ 5 │

julia> @evaluate @fuze df1 without df2
1×2 DataFrames.DataFrame
│ Row │ a │ b │
├─────┼───┼───┤
│ 1   │ 3 │ 6 │

julia> @evaluate begin
            df3 = DataFrame(c = [1, 2])
            @fuze df1 and df3
        end
6×3 DataFrames.DataFrame
│ Row │ a │ b │ c │
├─────┼───┼───┼───┤
│ 1   │ 1 │ 4 │ 1 │
│ 2   │ 1 │ 4 │ 2 │
│ 3   │ 2 │ 5 │ 1 │
│ 4   │ 2 │ 5 │ 2 │
│ 5   │ 3 │ 6 │ 1 │
│ 6   │ 3 │ 6 │ 2 │
```
"""
fuze(table1, conjunction, table2, on = intersect(names(table1), names(table2))) = begin
    word = get_expression(conjunction)
    if isempty(on)
        join(table1, table2, kind = :cross)
    else
        kind = MacroTools.@match word begin
            and => :inner
            or => :outer
            into => :right
            with => :semi
            without => :anti
        end
        join(table1, table2, on = on, kind = kind)
    end
end

"""
    @columns_from table selections...

Works by replacing each name with its order integer. This allows specifying
ranges as well as negative indices.

```jldoctest
julia> using LazyQuery

julia> @new_environment;

julia> @chain @evaluate begin
            using LazyQuery
            table = DataFrame(a = 1, b = 2, c = 3, d = 4, e = 5)
            @columns_from it a c:e
        end
1×4 DataFrames.DataFrame
│ Row │ a │ c │ d │ e │
├─────┼───┼───┼───┼───┤
│ 1   │ 1 │ 3 │ 4 │ 5 │

julia> @evaluate @columns_from table -a -(c:e)
1×1 DataFrames.DataFrame
│ Row │ b │
├─────┼───┤
│ 1   │ 2 │

julia> @evaluate @columns_from table a -b
ERROR: Cannot mix positive and negative indices
[...]
```
"""
columns_from(table, first_selection, other_selections...) = begin
    numbers = get_numbers(table, first_selection, other_selections)
    apply(x -> x[numbers], table)
end

"""
```jldoctest
julia> using LazyQuery

julia> @new_environment;

julia> @chain @evaluate begin
            using LazyQuery
            DataFrame(a = 1, b = 2)
            @rename it z = a y = b
        end
1×2 DataFrames.DataFrame
│ Row │ z │ y │
├─────┼───┼───┤
│ 1   │ 1 │ 2 │
```
"""
rename(table, args...) = begin
    f = x -> DataFrames.rename(x, Dict(reverse.(decompose_assignment.(get_expression.(args)))))
    apply(f, table)
end

"""
    @gather_from table selections...

```jldoctest
julia> using LazyQuery

julia> @new_environment;

julia> @chain @evaluate begin
            using LazyQuery
            DataFrame(a = 1, b = 2, c = 3, d = 4)
            @gather_from it a:b d
        end
3×3 DataFrames.DataFrame
│ Row │ variable │ value │ c │
├─────┼──────────┼───────┼───┤
│ 1   │ a        │ 1     │ 3 │
│ 2   │ b        │ 2     │ 3 │
│ 3   │ d        │ 4     │ 3 │
```
"""
gather_from(table, first_selection, other_selections...) = begin
    numbers = get_numbers(table, first_selection, other_selections)
    apply(x -> DataFrames.stack(table, numbers), table)
end

"""
    @spread_from table value *over* key

```jldoctest
julia> using LazyQuery

julia> @new_environment;

julia> @chain @evaluate begin
            using LazyQuery
            DataFrame(key = [:a, :b, :d], value = [1, 2, 4], c = 3)
            @spread_from it value over key
        end
1×4 DataFrames.DataFrame
│ Row │ c │ a │ b │ d │
├─────┼───┼───┼───┼───┤
│ 1   │ 3 │ 1 │ 2 │ 4 │
```
"""
spread_from(table, value, over_symbol, key) =
    DataFrames.unstack(table, get_expression(key), get_expression(value))
